/**
 * Copyright(c) 2013 Jade Techonologies Co., Ltd.
 *
 * History:
 *   14-8-4 下午12:21 Created by dwzhang
 */
package com.jade.framework.mom.impl;

import java.util.List;
import java.util.Map;

import javax.inject.Named;

import com.jade.framework.mom.*;

/**
 * Mom服务的实现。
 *
 * @author dwzhang
 * @version 1.0 14-8-4 下午12:21
 */
@Named ("system_momService")
public class MomServiceImpl implements MomService
{
    protected Map<String, MomProducer> producerMap;
    protected Map<String, MomConsumer> consumerMap;
    protected Map<String, MomBrowser> browserMap;

    public void setProducerMap(Map<String, MomProducer> producerMap)
    {
        this.producerMap = producerMap;
    }

    public void setConsumerMap(Map<String, MomConsumer> consumerMapMap)
    {
        this.consumerMap = consumerMapMap;
    }

    public void setBrowserMap(Map<String, MomBrowser> browserMapMap)
    {
        this.browserMap = browserMapMap;
    }

    protected MomProducer checkProducer(String topic)
            throws MomException
    {
        MomProducer producer = getProducer(topic);
        if (producer == null) {
            throw new MomException(MomConstants.ERR_NO_PRODUCER, "Producer " + topic + " not found", null,
                                   topic);
        }
        return producer;
    }

    protected MomConsumer checkConsumer(String topic)
            throws MomException
    {
        MomConsumer consumer = getConsumer(topic);
        if (consumer == null) {
            throw new MomException(MomConstants.ERR_NO_CONSUMER, "Consumer " + topic + " not found", null,
                                   topic);
        }
        return consumer;
    }

    protected MomBrowser checkBrowser(String topic)
            throws MomException
    {
        MomBrowser browser = getBrowser(topic);
        if (browser == null) {
            throw new MomException(MomConstants.ERR_NO_BROWSER, "Browser " + topic + " not found", null, topic);
        }
        return browser;
    }

    public MomSendResult send(String topic, String key, String tags, int delayLevel, Object msg)
            throws MomException
    {
        MomProducer producer = checkProducer(topic);
        return producer.send(topic, key, tags, delayLevel, msg);
    }

    public MomSendResult send(MomMessage message)
            throws MomException
    {
        MomProducer producer = checkProducer(message.getTopic());
        return producer.send(message);
    }

    public void send(MomMessage message, MomSendCallback callback)
            throws MomException
    {
        MomProducer producer = checkProducer(message.getTopic());
        producer.send(message, callback);
    }

    public MomMessage getMessage(String topic, String msgId)
            throws MomException
    {
        MomBrowser browser = checkBrowser(topic);
        return browser.getMessage(msgId);
    }

    public List<MomMessage> listMessage(String topic, String key, int maxNum)
            throws MomException
    {
        MomBrowser browser = checkBrowser(topic);
        return browser.listMessage(topic, key, maxNum);
    }

    public MomMessage getSingleMessage(String topic, String key)
            throws MomException
    {
        MomBrowser browser = checkBrowser(topic);
        return browser.getSingleMessage(topic, key);
    }

    public MomProducer getProducer(String topic)
    {
        return producerMap.get(topic);
    }

    public MomConsumer getConsumer(String topic)
    {
        return consumerMap.get(topic);
    }

    public void addProducer(String topic, MomProducer producer)
    {
        producerMap.put(topic, producer);
    }

    public void startProducer(String topic)
    {
        MomProducer producer = getProducer(topic);
        if (producer != null) {
            producer.start();
        }
    }

    public void stopProducer(String topic)
    {
        MomProducer producer = getProducer(topic);
        if (producer != null) {
            producer.stop();
        }
    }

    public void startConsumer(String topic)
    {
        MomConsumer consumer = getConsumer(topic);
        if (consumer != null) {
            consumer.start();
        }
    }

    public void stopConsumer(String topic)
    {
        MomConsumer consumer = getConsumer(topic);
        if (consumer != null) {
            consumer.stop();
        }
    }

    public void addConsumer(String topic, MomConsumer consumer)
    {
        consumerMap.put(topic, consumer);
    }

    public MomBrowser getBrowser(String topic)
    {
        return browserMap.get(topic);
    }
}
